package com.woniuxy.service.impl;

import com.woniuxy.po.Mail;
import com.woniuxy.po.TopicMail;
import com.woniuxy.service.Publisher;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: 马宇航
 * @Do: 处理 广播模式的service
 * @DateTime: 22/07/06/0006 下午 4:15
 * @Component: 成都蜗牛学苑
 **/
@Service
public class PublisherImpl implements Publisher {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Override
    public void fanoutMail(Mail mail) {
        rabbitTemplate.convertAndSend("fanout","",mail);
    }
    /**
     * 直连交换机消息发布
     * ChangeLog : 1. 创建 (22/07/06/0006 下午 5:03 [马宇航]);
     * @param topicMail
     * @return void
    */
    @Override
    public void directMail(TopicMail topicMail) {
        //设置响应回调，如果找不到交换机，则回调ack为false
        rabbitTemplate.setConfirmCallback(confirmCallback);
        //设置返回响应，发送到队列后（或者发送失败），则回调消息至returnsCallback,交换机到队列！
        rabbitTemplate.setReturnsCallback(returnsCallback);
        rabbitTemplate.convertAndSend("direct",topicMail.getRoutingkey(),topicMail);
    }
    //发送到正常的队列中
    @Override
    public void sendNormalQueue(Mail mail) {
        rabbitTemplate.convertAndSend("normalExchange","normalRouting",mail);
    }

    @Override
    public void sendTtlQueue(Mail mail) {
        rabbitTemplate.convertAndSend("normalExchange","ttlRouting",mail);
    }

    @Override
    public void sendDelayMessage(TopicMail topicMail) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        System.out.println("msg发送时间："+simpleDateFormat.format(new Date()));
        //id+时间戳  全局唯一 类似之前的邮戳 唯一
        CorrelationData correlationData = new CorrelationData(Math.random()+simpleDateFormat.format(new Date()));
        //设置响应回调，如果找不到交换机，则回调ack为false
        rabbitTemplate.setConfirmCallback(confirmCallback);
        //设置返回响应，发送到队列后（或者发送失败），则回调消息至returnsCallback,交换机到队列！
        rabbitTemplate.setReturnsCallback(returnsCallback);

        //发送消息，内部去封装消息头
        rabbitTemplate.convertAndSend("delayExchange", topicMail.getRoutingkey(), topicMail,
                new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        //设置消息头，添加x-delay头，对应就是消息延迟时间
                        double weight = topicMail.getWeight();
                        int delayTime = (int) weight;
                        //设置消息持久化
                        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                        //设置延迟具体的时间
                        //这个底层就是setHeader("x-delay",i);是一样的
                        message.getMessageProperties().setDelay(delayTime);
//                        message.getMessageProperties().setHeader("x-delay",delayTime);
                        return message;
                    }
                });
    }
    //解决交换机发送到队列 可能出问题的回调。
    final RabbitTemplate.ReturnsCallback returnsCallback = new RabbitTemplate.ReturnsCallback() {
        @Override
        public void returnedMessage(ReturnedMessage returned) {
            System.err.println("-------------------ReturnsCallback------------------------");
			System.out.println("ReturnedMessage,该参数具体业务具体使用 ："+returned.toString());
            System.err.println(new String(returned.getMessage().toString())+"消息被退回");
            System.err.println("消息退回原因："+returned.getReplyText());
            System.err.println("路由："+returned.getRoutingKey());
            System.err.println("获得消费者全局唯一标签："+returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation"));
        }
    };
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            //交换机回调给生产者的消息 如果交换机收到消息，返回true，否则
            if(ack){
                System.out.println("-------------------ConfirmCallback,发送到交换机成功------------------------");
                System.out.println("发送成功，进行业务操作....");
                if (correlationData!=null){
                    //交换机返回给生产者的消息
                    ReturnedMessage returned = correlationData.getReturned();
                    System.out.println("发送成功"+"correlationData全局id为："+correlationData.getId());
                }
            }else {
                System.out.println("-------------------ConfirmCallback,发送到交换机失败------------------------");
                if (correlationData!=null){
                    System.out.println("发送失败"+"correlationData全局id为："+correlationData.getId());
                }
                System.err.println("失败原因："+cause);
            }
        }
    };
}
